#include <stdlib.h>
#include <stdio.h>
#include <string.h>
#include <fcntl.h>
#include <errno.h>
#include <sys/types.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <netdb.h>
#include <unistd.h>
#include <err.h>
#include <libmscommon/mscommon.h>
#include <libmscommon/msnetwork.h>
#include <libmscommon/msepoll.h>
#include <libmscommon/mstime.h>
#include <libmscommon/msmd.h>
#include <libmscommon/msenv.h>
#include <libmscommon/msmem.h>
#include <libmslog/mslog.h>
#include <libmsprotocol/msprotocol.h>

#define FLAG "PTL_TEST"

#define mshostname_ipaddr	"192.168.200.244"
#define mshostname_port	50000
#define mshostname_url_http	"http://127.0.0.1:8010/ch31.ts"
#define mshostname_url_rtsp	"rtsp://127.0.0.1:8010/ch31.ts"

typedef enum{
	TESTMODE_HTTP=0x00,
	TESTMODE_RTSPPULL_OVER_UDP,	
	TESTMODE_RTSP_OVER_TCP,
	TESTMODE_RTSPPUSH_OVER_UDP,	
	TESTMODE_UDP,	
	TESTMODE_MSAV,	
	TESTMODE_MSOSHLS,	
	TESTMODE_CONNECTTEST,	
	TESTMODE_ALL
}TESTMODE;


typedef struct TESTFUNCContex{
	ms_byte name[64];
	TESTMODE testmode;
	ms_u32 total_loop;
	ms_bool flag_iotest;
	ms_bool flag_302return;
	URLContext *pcilent_ctt_list;
	URLContext msusrctt_out;
	URLProtocol *msptl_in;
	URLProtocol *msptl_out;
	
	ms_s32 thread_index;
	
	ms_s32 test_index_start;
	ms_s32 test_index_end;
}TESTFUNCContex;

void testfunc_open(TESTFUNCContex * ptestfunc_txt,ms_bool flag_out)
{
	ms_s32 index;
	URLContext *pcilent_ctt=ms_null;
	for(index=ptestfunc_txt->test_index_start;index<ptestfunc_txt->test_index_end;index++){ 
		pcilent_ctt=&ptestfunc_txt->pcilent_ctt_list[index];
		if(ms_true==flag_out){
			pcilent_ctt->flags|=FLAG_WRITE;
		}else{
			if(0!=ptestfunc_txt->flag_302return){
				pcilent_ctt->opt.http.flag_302return=ms_true;
			}
			pcilent_ctt->flag_location=ms_false;
		}
		ptestfunc_txt->msptl_in->url_open(pcilent_ctt);
	}
}
void testfunc_send(TESTFUNCContex * ptestfunc_txt)
{
	ms_s32 index;
	URLContext *pcilent_ctt=ms_null;
	ms_byte * strHe="Hello world!!!!!!!!!!!!!!!!!!欢迎使用MSCORE，它的全名为Multi-Core-Server，即多核处理服务器，设计时主要考虑多核并行处理，项目创建于2015年6月4日，功能结构会不断的更新优化。代码设计围绕“简洁高效，模块化设计，快速扩展，具有较好的移植性”的基本原则，在负载吞吐方面尽最大可能达到良好的系统和物理可扩展性。MSCORE是以“点对点”模型（即主动下拉和上推模型）作为原始设计框架，而非并发C/S服务模型。当然，这并不是说MSCORE不关注并发服务，而是使用MSCORE服务也能较轻松的衍生出大并发的C/S架构，如rtmp，hls，http等主流并发流，来提供完整的解决方案。MSCORE主要是用于解决各种网络流及本地媒体文件的传输，储存，加密，流质量检测，流质量监测，流内容或音视频内容更改的解决方案。它可广泛应用于军事，科技探索，广电，新闻，医院，酒店，安防，学校，网络直播，甚至于个人住所等领域，总而言之 HTTP/1.0";
	for(index=ptestfunc_txt->test_index_start;index<ptestfunc_txt->test_index_end;index++){ 
		pcilent_ctt=&ptestfunc_txt->pcilent_ctt_list[index];
		ptestfunc_txt->msptl_in->url_write(pcilent_ctt,strHe,ms_buflen(strHe));
	}
}

void testfunc_io(TESTFUNCContex * ptestfunc_txt)
{
	ms_s32 index;
	URLContext *pcilent_ctt=ms_null;
	ms_u08 recvbuf[mscfg_maxlen_recv]={0};	/* buffer to receive data */
	ms_s32 numTest=ptestfunc_txt->test_index_end-ptestfunc_txt->test_index_start+1;
	ms_u64 curTime=mstime_api_us();
	ms_u64 baseTimeUs[(const ms_s32)numTest];
	ms_s32 indexBaseTime;
	while(1){
		for(index=ptestfunc_txt->test_index_start;index<ptestfunc_txt->test_index_end;index++){ 
			indexBaseTime=index-ptestfunc_txt->test_index_start;
			pcilent_ctt=&ptestfunc_txt->pcilent_ctt_list[index];
			ms_s32 len=ptestfunc_txt->msptl_in->url_read( pcilent_ctt, recvbuf, mscfg_maxlen_recv);
			if(len>0){
				baseTimeUs[indexBaseTime]=mstime_api_us();
				if(0==index){
					ptestfunc_txt->msptl_out->url_write(&ptestfunc_txt->msusrctt_out, recvbuf, len);
				}
			}else if(-3==len){
				ptestfunc_txt->msptl_in->url_close(pcilent_ctt);
				if(0!=ptestfunc_txt->flag_302return){
					pcilent_ctt->opt.http.flag_302return=ms_true;
				}
				pcilent_ctt->flag_location=ms_false;
				ptestfunc_txt->msptl_in->url_open(pcilent_ctt);
			}else{
				if(mstime_api_counterAsyncUs(&baseTimeUs[indexBaseTime], ms_usseconds(10))){
					ptestfunc_txt->msptl_in->url_close(pcilent_ctt);
					ptestfunc_txt->msptl_in->url_open(pcilent_ctt);
				}else{
					ms_usleep(50);
				}
			}
		}
	}
}
void testfunc_close(TESTFUNCContex * ptestfunc_txt)
{
	ms_s32 index;
	URLContext *pcilent_ctt=ms_null;
	for(index=ptestfunc_txt->test_index_start;index<ptestfunc_txt->test_index_end;index++){ 
		pcilent_ctt=&ptestfunc_txt->pcilent_ctt_list[index];
		ptestfunc_txt->msptl_in->url_close(pcilent_ctt);
	}
}
static ms_pvoid  test_func_noread(ms_void * ms_in arglist)
{
	TESTFUNCContex * ptestfunc_cxt=(TESTFUNCContex * )arglist;
	ms_s32 index;
	ms_info("[%s]---test_index(%d:%d)",ptestfunc_cxt->name,ptestfunc_cxt->test_index_start,ptestfunc_cxt->test_index_end);
ms_info("[%s]BN------------------------",ptestfunc_cxt->name);
		ms_info("[%s]OPEN BN------------------------",ptestfunc_cxt->name);
		if(TESTMODE_RTSPPUSH_OVER_UDP==ptestfunc_cxt->testmode){
			testfunc_open(ptestfunc_cxt,ms_true);
		}else{
			testfunc_open(ptestfunc_cxt,ms_false);
		}
		ms_info("[%s]OPEN EN------------------------",ptestfunc_cxt->name);
ms_info("[%s]EN------------------------",ptestfunc_cxt->name);
}
static ms_pvoid  test_func_nor(ms_void * ms_in arglist)
{
	TESTFUNCContex * ptestfunc_cxt=(TESTFUNCContex * )arglist;
	ms_s32 index;
	ms_info("[%s]---test_index(%d:%d)",ptestfunc_cxt->name,ptestfunc_cxt->test_index_start,ptestfunc_cxt->test_index_end);
	while(1)
	{
ms_info("[%s]BN------------------------",ptestfunc_cxt->name);
		ms_info("[%s]OPEN BN------------------------",ptestfunc_cxt->name);
		if(TESTMODE_RTSPPUSH_OVER_UDP==ptestfunc_cxt->testmode){
			testfunc_open(ptestfunc_cxt,ms_true);
		}else{
			testfunc_open(ptestfunc_cxt,ms_false);
		}
		ms_info("[%s]OPEN EN------------------------",ptestfunc_cxt->name);
		if(TESTMODE_CONNECTTEST==ptestfunc_cxt->testmode){
			testfunc_send(ptestfunc_cxt);
		}
		if(ms_true==ptestfunc_cxt->flag_iotest){
			testfunc_io(ptestfunc_cxt);
			ms_sleep(1);
		}else{
			ms_sleep(1);
		}
		
		ms_info("[%s]CLOSE BN------------------------",ptestfunc_cxt->name);
		testfunc_close(ptestfunc_cxt);
		ms_info("[%s]CLOSE EN------------------------",ptestfunc_cxt->name);

ms_info("[%s]EN------------------------",ptestfunc_cxt->name);
		
		ms_sleep(3);
		if((ptestfunc_cxt->total_loop>0) && ms_lessThan((--ptestfunc_cxt->total_loop), 1)){
			break;
		}
	}
}

static ms_pvoid  test_func(ms_void * ms_in arglist)
{
	TESTFUNCContex * ptestfunc_cxt=(TESTFUNCContex * )arglist;
#if 0	
	if(ms_true==ptestfunc_cxt->flag_iotest){
		if(1==ptestfunc_cxt->thread_index){
			test_func_noread(arglist);
		}else{
			test_func_nor(arglist);
		}
	}else{
		test_func_nor(arglist);
	}
#else
	test_func_nor(arglist);
#endif
}


int main(int argc, char **argv)
{
	mslog_api_init( 53, ms_null,ms_null, ms_null);
	if(argc==1){
		argv[1]="0";
		argv[2]="http://127.0.0.1:8010/ch1";
		argv[3]="1000";
		argv[4]="10";
		argv[5]="0";
		argv[6]="1";
		argv[7]="1";
		argv[8]="52";
		argc=9;
	}

	if(argc<8){
		ms_error("Userage:%s mode(0-http,1-rtsp_pull(udp),2-rtsp_pull(tcp),3-rtsp_push(udp),4-udp,5-msav,6-msoshls,7-connect_test)   urlin urlout_listen  total_connect  total_loop 302return  iotest  num_thread [logopt ] " ,argv[0]);
		return 0;
	}
	msthread_api_envInit();
	msthread_api_setName("mscore_loadtool");
	msmem_api_init(ms_false);
	msptc_api_init();
	msptc_api_register();
	msmd_api_setUlimit();
	mssignal_api_catch("loadtool");
	
	ms_debug("Get param");
	TESTMODE testmode=ms_atoi(argv[1]);
	ms_byte urlin[1024]={0};
	ms_strcpy(urlin, (argv[2]));
	ms_byte urlout_listen[1024]={0};
	ms_strcpy(urlout_listen, (argv[3]));
	ms_s32 total_connect=ms_atoi(argv[4]);
	ms_s32 total_loop=ms_atoi(argv[5]);
	ms_bool flag_302return=ms_atoi(argv[6]) ;
	ms_bool flag_iotest=ms_atoi(argv[7]) ;
	ms_u32 num_thread=ms_atoi(argv[8]) ;
	ms_s32 logopt=(10==argc) ? ms_atoi(argv[9]) : 51;
	
	mslog_api_init( logopt, ms_null,ms_null, ms_null);
	ms_info("total_connect:%d,total_loop:%d,flag_302return:%d,num_thread:%d", total_connect,total_loop,flag_302return,num_thread);
	
	URLContext *pcilent_ctt_list=ms_null;
	ms_malloc_retErr(-1,pcilent_ctt_list,(total_connect*sizeof(URLContext)),"pcilent_ctt_list");
	ms_s32 index=0;
	URLContext *pcilent_ctt=ms_null;

	ms_debug("pcilent_ctt_list:%#x",pcilent_ctt_list);
//Get urlin	
	for(index=0;index<total_connect;index++){ 
		pcilent_ctt=&pcilent_ctt_list[index];
		ms_strcpy(pcilent_ctt->url,urlin);
	}
//Get msptl_in	
	URLProtocol *msptl_in=ms_null;
	switch(testmode){
		case TESTMODE_HTTP:
			msptl_in=msptc_api_matchByPtcindex(msptc_http);
			break;
		case TESTMODE_RTSPPULL_OVER_UDP:
			msenv_api_set(msenv_ptl_rtsptransport,"udp",ms_true);
			msptl_in=msptc_api_matchByPtcindex(msptc_rtsp);
			break;	
		case TESTMODE_RTSP_OVER_TCP:
			msenv_api_set(msenv_ptl_rtsptransport,"tcp",ms_true);
			msptl_in=msptc_api_matchByPtcindex(msptc_rtsp);
			break;
		case TESTMODE_RTSPPUSH_OVER_UDP:
			msenv_api_set(msenv_ptl_rtsptransport,"udp",ms_true);
			for(index=0;index<total_connect;index++){ 
				pcilent_ctt=&pcilent_ctt_list[index];
				ms_sprintfs(pcilent_ctt->url,"rtsp://127.0.0.1:8010/designated_ch%d",(index+1));
			}
			msptl_in=msptc_api_matchByPtcindex(msptc_rtsp);
			break;	
		case TESTMODE_UDP:
			msptl_in=msptc_api_matchByPtcindex(msptc_udp);
			URLProtocol *msptl_out=msptc_api_matchByPtcindex(msptc_udp);
			URLContext msusrctt_in;
			URLContext msusrctt_out;
			ms_memset(&msusrctt_in,0,sizeof(URLContext));
			ms_memset(&msusrctt_out,0,sizeof(URLContext));
			ms_strcpy(msusrctt_in.url, "udp://127.0.0.1:3006");
			ms_strcpy(msusrctt_out.url, "udp://224.2.2.2:60000");
			msusrctt_out.flags=FLAG_WRITE;
			msusrctt_out.opt.udp.ifcOut="eth0";
			msusrctt_out.opt.udp.ipOut="192.168.59.136";
			msptl_in->url_open(&msusrctt_in);
			msptl_out->url_open(&msusrctt_out);
			ms_byte buf[5120]={0};
			while(1){
				int recvlen=msptl_in->url_read( &msusrctt_in, buf , 5120 );
				ms_usleep(100);
				int sendlen=msptl_out->url_write( &msusrctt_out, buf , recvlen );
			}
			break;	
		case TESTMODE_MSAV:
			msptl_in=msptc_api_matchByPtcindex(msptc_msav);
			break;
		case TESTMODE_MSOSHLS:
			msptl_in=msptc_api_matchByPtcindex(msptc_msoshls);
			break;
		case TESTMODE_CONNECTTEST:
			msptl_in=msptc_api_matchByPtcindex(msptc_tcp);
			break;
		case TESTMODE_ALL:
			break;
	}

	TESTFUNCContex  testfunc_cxt_list[512];
	MSTHREADContext msthread_ctt_list[512];
	ms_buf0(testfunc_cxt_list);
	ms_buf0(msthread_ctt_list);
	ms_info("--------num_thread:%d",num_thread);
	for(index=0;index<num_thread;index++){ 
		TESTFUNCContex  *ptestfunc_cxt=&testfunc_cxt_list[index];
		ms_s32 num_bei=(total_connect/num_thread);
		if(0==num_bei){
			num_bei=1;
		}
		ptestfunc_cxt->test_index_start=ms_min(index*num_bei, total_connect);
		if(index==(num_thread-1)){	//13 ->[(13/3)*2]:13
			ptestfunc_cxt->test_index_end=total_connect;
		}else{
			ptestfunc_cxt->test_index_end=ms_min((index+1)*num_bei, total_connect);
		}
		ms_sprintfs(ptestfunc_cxt->name, "test_fuc%d", index);
		ptestfunc_cxt->thread_index		=index;
		ptestfunc_cxt->testmode			=testmode;
		ptestfunc_cxt->total_loop		=total_loop;
		ptestfunc_cxt->flag_iotest		=flag_iotest;
		ptestfunc_cxt->flag_302return		=flag_302return;
		ptestfunc_cxt->pcilent_ctt_list		=pcilent_ctt_list;
		ptestfunc_cxt->msptl_in			=msptl_in;
		if(0==index){
			ptestfunc_cxt->msptl_out=msptc_api_matchByPtcindex(msptc_udp);
			ms_memset(&ptestfunc_cxt->msusrctt_out,0,sizeof(URLContext));
			ms_strcpy(ptestfunc_cxt->msusrctt_out.url, urlout_listen);
			ptestfunc_cxt->msusrctt_out.flags=FLAG_WRITE;
			ptestfunc_cxt->msptl_out->url_open(&ptestfunc_cxt->msusrctt_out);
		}

		MSTHREADContext  *pmsthread_ctt=&msthread_ctt_list[index];
		msthread_api_create(pmsthread_ctt, ptestfunc_cxt->name, test_func, ptestfunc_cxt);
	}
	while(1){
		ms_sleep(10);
	}
end:	
	ms_deMalloc(pcilent_ctt_list);
	msthread_api_envDeinit();
	msptc_api_deinit();
	msmem_api_deinit();
	mslog_api_deinit();
}
#undef MAIN_C

